package com.atguigu.flink.day01;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.operators.AggregateOperator;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.FlatMapOperator;
import org.apache.flink.api.java.operators.UnsortedGrouping;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

/**
 * @author Felix
 * @date 2024/8/09
 * 以批的形式对有界数据进行处理
 */
public class Flink01_WC_Bound_Batch {
    public static void main(String[] args) {
        //TODO 1.准备环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        //TODO 2.从文件中读取数据
        DataSource<String> ds = env.readTextFile("D:\\dev\\workspace\\bigdata-0318\\input\\words.txt");
        //TODO 3.对读取的数据进行扁平化处理  封装为二元组
        FlatMapOperator<String, Tuple2<String, Long>> flapMapDS = ds.flatMap(
                new FlatMapFunction<String, Tuple2<String, Long>>() {
                    @Override
                    public void flatMap(String lineStr, Collector<Tuple2<String, Long>> out) throws Exception {
                        //使用分隔符对读取到的一行数据进行分隔
                        String[] wordArr = lineStr.split(" ");
                        //遍历数组获取每一个单词
                        for (String word : wordArr) {
                            //将单词封装为二元组并向下游传递
                            out.collect(Tuple2.of(word, 1L));
                        }
                    }
                }
        );
        //TODO 4.按照单词进行分组
        UnsortedGrouping<Tuple2<String, Long>> groupDS = flapMapDS.groupBy(0);
        //TODO 5.聚合计算
        AggregateOperator<Tuple2<String, Long>> sumDS = groupDS.sum(1);
        //TODO 6.将结果进行打印输出
        try {
            //(flink,1)
            //(world,1)
            //(hello,3)
            //(java,1)
            sumDS.print();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

    }
}
